/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package consys.social.core.dao;

import org.apache.log4j.Logger;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.util.Bytes;
import java.util.Map;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Get;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.hbase.client.Put;
import java.io.IOException;
import consys.social.core.bo.Entity;
import consys.social.nosql.dao.hbase.HBase;
import consys.social.memcached.dao.spymemcached.Memcached;
import java.util.Random;
import static consys.social.nosql.service.hbase.HBaseHelpers.*;

/**
 *
 * @author petan
 */
public class FollowDaoImpl implements FollowDao {

    private Memcached memcached;
    private ThreadLocal<HBase> entities;
    private Logger logger = Logger.getLogger(FollowDaoImpl.class);

    public void setEntities(ThreadLocal<HBase> entities) {
        this.entities = entities;
    }

    public void setMemcached(Memcached memcached) {
        this.memcached = memcached;
    }

    @Override
    public void follow(Entity who, Entity whom) throws IOException {
        if (memcached == null) {
            throw new IllegalStateException("Memcached not initialized");
        }
        if (entities.get() == null) {
            throw new IllegalStateException("HBase not initialized");
        }
        try {
            Get get = createGet(whom.getUid());
            addToGet(get, "blocked", who.getUid());
            if (!entities.get().exists(get)) {
                Put put = createPut(who.getUid());
                addToPut(put, "following", whom.getUid(), true);
                logger.info("Adding following into HBase");
                if (entities.get().add(put)) {
                    long value = entities.get().increment(who.getUid(), "count", "following", 1);
                    memcached.set(who.getUid() + ":following-count", value, 0);
                }
                put = createPut(whom.getUid());
                addToPut(put, "followers", who.getUid(), true);
                if (entities.get().add(put)) {
                    long value = entities.get().increment(whom.getUid(), "count", "followers", 1);
                    memcached.set(whom.getUid() + ":followers-count", value, 0);
                    List<String> followers = getAllFollowers(whom);
                    if (!followers.contains(who.getUid())) {
                        followers.add(who.getUid());
                    }
                    memcached.set(whom.getUid() + ":followers", followers, 0);
                }
            }
        } catch (IOException ex) {
            logger.error("Follow problem in HBase");
            throw ex;
        }
    }

    @Override
    public void unfollow(Entity who, Entity whom) throws IOException {
        if (memcached == null) {
            throw new IllegalStateException("Memcached not initialized");
        }
        if (entities.get() == null) {
            throw new IllegalStateException("HBase not initialized");
        }
        try {
            Delete delete = createDelete(who.getUid());
            deleteCol(delete, "following", whom.getUid());
            logger.info("Unfollowing in HBase");
            if (entities.get().deleteIfTrue(delete)) {
                long value = entities.get().increment(who.getUid(), "count", "following", -1);
                memcached.set(who.getUid() + ":following-count", value, 0);
            }
            delete = createDelete(whom.getUid());
            deleteCol(delete, "followers", who.getUid());
            if (entities.get().deleteIfTrue(delete)) {
                long value = entities.get().increment(whom.getUid(), "count", "followers", -1);
                memcached.set(whom.getUid() + ":followers-count", value, 0);
                List<String> followers = getAllFollowers(whom);
                followers.remove(who.getUid());
                memcached.set(whom.getUid() + ":followers", followers, 0);
            }
        } catch (IOException ex) {
            logger.error("Unfollow problem in HBase");
            throw ex;
        }
    }

    @Override
    public void block(Entity follower, Entity leader) throws IOException {
        unfollow(follower, leader);
        Put put = createPut(leader.getUid());
        addToPut(put, "blocked", follower.getUid(), true);
        entities.get().add(put);
    }

    @Override
    public List<String> getAllFollowers(Entity entity) throws IOException {
        List<String> followers = (List<String>) memcached.get(entity.getUid() + ":followers");
        logger.info("Getting followers");
        if (followers == null) {
            followers = new ArrayList<String>();
            Get get = createGet(entity.getUid());
            addFamilyToGet(get, "followers");
            Result res = null;
            try {
                res = entities.get().get(get);
            } catch (IOException ex) {
                logger.error("HBase and Memcached probably down");
                throw ex;
            }
            Map<byte[], byte[]> results = res.getFamilyMap("followers".getBytes());
            if (results != null) {
                for (byte[] oneFollower : results.keySet()) {
                    followers.add(Bytes.toString(oneFollower));
                }
            }
            memcached.set(entity.getUid() + ":followers", followers, 0);
        }
        return followers;
    }

    @Override
    public List<String> getAllFollowings(Entity entity) throws IOException {
        List<String> followings = new ArrayList<String>();
        Get get = createGet(entity.getUid());
        addFamilyToGet(get, "following");
        Result res = null;
        logger.info("Getting all followings");
        try {
            res = entities.get().get(get);
        } catch (IOException ex) {
            logger.error("Getting all followings problem");
            throw ex;
        }
        if (res != null) {
            Map<byte[], byte[]> results = res.getFamilyMap("following".getBytes());
            if (results != null) {
                for (byte[] oneFollower : results.keySet()) {
                    followings.add(Bytes.toString(oneFollower));
                }
            }
        }
        return followings;
    }

    @Override
    public int getFollowersCount(Entity who) throws IOException {
        return getCount(who, true);
    }

    @Override
    public int getFollowingCount(Entity who) throws IOException {
        return getCount(who, false);
    }

    private int getCount(Entity who, boolean followers) throws IOException {
        String follow = followers ? "followers" : "following";
        logger.info("Getting " + followers + " count");
        Integer count = (Integer) memcached.get(who.getUid() + ":" + follow + "-" + "count");
        if (count == null) {
            Get g = createGet(who.getUid());
            addToGet(g, "count", follow);
            try {
                Result r = entities.get().get(g);
                count = Bytes.toInt(r.value());
                memcached.set(who.getUid() + ":" + follow + "-" + "count", count, 0);
            } catch (IOException ex) {
                logger.error("Getting " + follow + " count error");
                throw ex;
            }
        }
        return count;
    }

    @Override
    public List<Entity> getFollowings(Entity entity, int from, int howMany) throws IOException {
        logger.info("Getting followings");
        List<String> followings = getAllFollowings(entity);
        return getFollows(followings, from, howMany);
    }

    @Override
    public List<Entity> getFollowers(Entity entity, int from, int howMany) throws IOException {
        logger.info("Getting followers");
        List<String> followers = getAllFollowers(entity);
        return getFollows(followers, from, howMany);
    }

    private List<Entity> getFollows(List<String> flws, int from, int howMany) {
        List<Entity> flwEntities = new ArrayList<Entity>();
        for (int i = from; i < Math.min(from + howMany, flws.size()); i++) {
            flwEntities.add(new Entity(flws.get(i)));
        }
        return flwEntities;
    }

    @Override
    public List<Entity> getRandomFollowings(Entity entity, int count) throws IOException {
        logger.info("Getting random followings");
        List<String> followings = getAllFollowings(entity);
        return chooseRandomEntities(followings, count);
    }

    @Override
    public List<Entity> getRandomFollowers(Entity entity, int count) throws IOException {
        logger.info("Getting random followers");
        List<String> followers = getAllFollowers(entity);
        return chooseRandomEntities(followers, count);
    }

    private List<Entity> chooseRandomEntities(List<String> uids, int count) {
        List<Entity> flwEntities = new ArrayList<Entity>();
        if (count < uids.size()) {
            Random randomizer = new Random();
            for (int i = 0; i < count; i++) {
                int randomUid = randomizer.nextInt(uids.size());
                flwEntities.add(new Entity(uids.get(randomUid)));
                uids.remove(randomUid);
            }
        } else {
            for (String uid : uids) {
                flwEntities.add(new Entity(uid));
            }
        }
        return flwEntities;
    }

    @Override
    public boolean isFollowing(Entity who, Entity whom) throws IOException {
        return getAllFollowers(whom).contains(who.getUid());
    }
}
